/*
 * $AIST_Release: 0.9.0 $
 * Copyright 2011 Information Technology Research Institute, National
 * Institute of Advanced Industrial Science and Technology
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *    http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import org.scalatest.FunSuite

import java.io.File
import java.io.FileOutputStream
import com.google.protobuf.Descriptors.FieldDescriptor.Type
import com.google.protobuf.DescriptorProtos._
import com.google.protobuf.Descriptors._
import com.google.protobuf.ByteString
import com.google.protobuf.CodedOutputStream
import com.google.protobuf.DynamicMessage
import aggr._

import Aggregators._
import Serializer._
import Deserializer._
import pbdata.Tables.KeyValue
import util.JavaTypes._

class AggrSuite extends FunSuite {

  def useTempFile(data: List[(ByteString, ByteString)])(f: String => Unit) = {
    val tmp = File.createTempFile("scdata", null)
    try {
      val os  = new FileOutputStream(tmp)
      val cos = CodedOutputStream.newInstance(os)

      try {
        for ((k, v) <- data) {
          cos.writeMessage(1, KeyValue.newBuilder().setKey(k).setValue(v).build())
        }
        cos.flush()
        os.close()

        f(tmp.toString)
      } finally {
        os.close()
      }
    } finally {
      tmp.delete()
    }
  }

  test("Init test") {
    println(Aggregators.aggrMap)
  }

  test("Kick test") {
    val data = List(
        serializeInt64(1) -> serializeInt64(1),
        serializeInt64(2) -> serializeInt64(2),
        serializeInt64(3) -> serializeInt64(3),
        serializeInt64(1) -> serializeInt64(4),
        serializeInt64(2) -> serializeInt64(5),
        serializeInt64(3) -> serializeInt64(6),
        serializeInt64(1) -> serializeInt64(7),
        serializeInt64(2) -> serializeInt64(8),
        serializeInt64(3) -> serializeInt64(9))

    useTempFile(data) { fn =>
      var p = kick("/dev/stdout",
        "collection",
        SerializedTypes.Int64Type,
        SerializedTypes.Int64Type,
        List(),
        fn, true)

      val c = p.waitFor()
      assert(c == 0, "Exit code is not 0.")
    }
  }

  test("Sum test") {
       
    val data = List(
        serializeString("hoge") -> serializeInt64(1),
        serializeString("moke") -> serializeInt64(2),
        serializeString("kuma") -> serializeInt64(3),
        serializeString("kuma") -> serializeInt64(4),
        serializeString("hoge") -> serializeInt64(5),
        serializeString("moke") -> serializeInt64(6),
        serializeString("kuma") -> serializeInt64(7),
        serializeString("moke") -> serializeInt64(8),
        serializeString("hoge") -> serializeInt64(9))

    useTempFile(data) { fn =>
      var p = kick(
        "/dev/stdout",
        "sum",
        SerializedTypes.StringType,
        SerializedTypes.Int64Type,
        List(),
        fn, true)

      val c = p.waitFor()
      assert(c == 0, "Exit code is not 0.")
    }
  }

  test("Sum test2") {

    val fdp = FileDescriptorProto.newBuilder().addMessageType(
      DescriptorProto.newBuilder()
        .setName("message1")
        .addField(FieldDescriptorProto.newBuilder()
          .setLabel(FieldDescriptorProto.Label.LABEL_REQUIRED)
          .setType(FieldDescriptorProto.Type.TYPE_INT64)
          .setName("val1")
          .setNumber(1))
        .addField(FieldDescriptorProto.newBuilder()
          .setLabel(FieldDescriptorProto.Label.LABEL_REQUIRED)
          .setType(FieldDescriptorProto.Type.TYPE_INT64)
          .setName("val2")
          .setNumber(2))).build()

    val fd = FileDescriptor.buildFrom(fdp, Array())

    val desc = fd.findMessageTypeByName("message1")
    val val1Desc = desc.findFieldByName("val1")
    val val2Desc = desc.findFieldByName("val2")
    val valType = SerializedTypes.MessageType(desc)

    def mkMes(val1: Long, val2: Long): ByteString = {
      val m = DynamicMessage.newBuilder(desc)
        .setField(val1Desc, new JLong(val1))
        .setField(val2Desc, new JLong(val2))
        .build()
      m.toByteString()
    }
       
    val data = List(
        serializeString("hoge") -> mkMes(1, 2),
        serializeString("moke") -> mkMes(10, 20),
        serializeString("kuma") -> mkMes(43, 42),
        serializeString("kuma") -> mkMes(25, 212),
        serializeString("hoge") -> mkMes(394, 438),
        serializeString("moke") -> mkMes(348, 348),
        serializeString("kuma") -> mkMes(112, 1024),
        serializeString("moke") -> mkMes(104, 474),
        serializeString("hoge") -> mkMes(999, 1))

    useTempFile(data) { fn =>
      var p = kick(
        "/dev/stdout",
        "sum",
        SerializedTypes.StringType,
        valType,
        List(),
        fn, true)

      val c = p.waitFor()
        assert(c == 0, "Exit code is not 0.")
    }
  }

  test("Serializer/Deserializer test") {

    def roundtrip[T](seri: (T => ByteString), deseri: (ByteString => T), value: T) = {
      assert(deseri(seri(value)) == value)
    }

    roundtrip(serializeBool,     deserializeBool,     true)
    roundtrip(serializeDouble,   deserializeDouble,   11.3)
    roundtrip(serializeEnum,     deserializeEnum,     10)
    roundtrip(serializeFixed32,  deserializeFixed32 , 32)
    roundtrip(serializeFixed64,  deserializeFixed64 , 64l)
    roundtrip(serializeSFixed32, deserializeSFixed32, 32)
    roundtrip(serializeSFixed64, deserializeSFixed64, 64l)
    roundtrip(serializeFloat,    deserializeFloat,    11.1f)
    roundtrip(serializeInt32,    deserializeInt32,    111)
    roundtrip(serializeInt64,    deserializeInt64,    3543l)
    roundtrip(serializeSInt32,   deserializeSInt32,   111)
    roundtrip(serializeSInt64,   deserializeSInt64,   35323l)
    roundtrip(serializeUInt32,   deserializeUInt32,   111)
    roundtrip(serializeUInt64,   deserializeUInt64,   35323l)
    roundtrip(serializeString,   deserializeString,   "hogehogehoge")
  }
}
